import com.alibaba.fastjson2.JSONArray;
import com.alibaba.fastjson2.JSONObject;
import com.arcadedb.database.Database;
import com.arcadedb.exception.CommandSQLParsingException;
import com.arcadedb.graph.MutableVertex;
import com.arcadedb.graph.Vertex;
import com.arcadedb.query.sql.executor.ResultSet;
import com.arcadedb.schema.Property;
import com.arcadedb.schema.Schema;
import com.arcadedb.schema.Type;
import com.arcadedb.schema.VertexType;
import exception.*;
import indi.hjhk.exception.ExceptionSerializer;
import indi.hjhk.log.Logger;
import nju.hjh.arcadedb.timeseries.DataPointSet;
import nju.hjh.arcadedb.timeseries.DataType;
import nju.hjh.arcadedb.timeseries.TSUpdateStrategy;
import nju.hjh.arcadedb.timeseries.TimeseriesEngine;
import nju.hjh.arcadedb.timeseries.datapoint.DataPoint;
import nju.hjh.arcadedb.timeseries.datapoint.DoubleDataPoint;
import nju.hjh.arcadedb.timeseries.datapoint.LongDataPoint;
import nju.hjh.arcadedb.timeseries.datapoint.StringDataPoint;
import nju.hjh.arcadedb.timeseries.exception.TimeseriesException;
import nju.hjh.arcadedb.timeseries.statistics.FixedStatistics;
import nju.hjh.arcadedb.timeseries.statistics.NumericStatistics;
import nju.hjh.arcadedb.timeseries.statistics.Statistics;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.lang.reflect.Constructor;
import java.math.BigDecimal;
import java.net.InetAddress;
import java.net.Socket;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;

public class ArcadeTSDBWorker implements Runnable{
    public static final String TYPE_OBJECT = ArcadeTSDBServer.TYPE_OBJECT;
    public static final String PROP_OBJECT_ID = ArcadeTSDBServer.PROP_OBJECT_ID;
    public static final int SHOW_MESSAGE_LENGTH = ArcadeTSDBServer.SHOW_MESSAGE_LENGTH;
    public static final ArrayList<String> METRIC_QUERY_TYPE = new ArrayList<>();

    static {
        METRIC_QUERY_TYPE.add("all");
        METRIC_QUERY_TYPE.add("first");
        METRIC_QUERY_TYPE.add("last");
        METRIC_QUERY_TYPE.add("count");
        METRIC_QUERY_TYPE.add("max");
        METRIC_QUERY_TYPE.add("min");
        METRIC_QUERY_TYPE.add("sum");
        METRIC_QUERY_TYPE.add("avg");
    }

    public Socket socket;
    public final InetAddress remoteAddress;
    public final int remotePort;
    public final Logger logger;

    public static Vertex getOrNewVertexById(Database database, String vertexType, String idKey, String idValue){
        if (!database.getSchema().existsType(vertexType)){
            VertexType type = database.getSchema().createVertexType(vertexType);
            Property keyProp = type.createProperty(idKey, Type.STRING);
            keyProp.createIndex(Schema.INDEX_TYPE.LSM_TREE, true);
        }

        ResultSet rs = database.query("SQL", String.format("SELECT FROM %s WHERE %s='%s'", vertexType, idKey, idValue));
        if (rs.hasNext()){
            return rs.next().getVertex().orElseGet(() -> {
                MutableVertex newVertex = database.newVertex(vertexType);
                newVertex.set(idKey, idValue);
                newVertex.save();
                return newVertex;
            });
        }else{
            MutableVertex newVertex = database.newVertex(vertexType);
            newVertex.set(idKey, idValue);
            newVertex.save();
            return newVertex;
        }
    }

    public static Vertex getVertexById(Database database, String vertexType, String idKey, String idValue){
        if (!database.getSchema().existsType(vertexType))
            return null;

        ResultSet rs = database.query("SQL", String.format("SELECT FROM %s WHERE %s='%s'", vertexType, idKey, idValue));
        if (rs.hasNext()){
            return rs.next().getVertex().orElseGet(() -> null);
        }else{
            return null;
        }
    }

    public ArcadeTSDBWorker(Socket socket, Logger serverLogger) {
        this.socket = socket;
        remoteAddress = socket.getInetAddress();
        remotePort = socket.getPort();
        logger = serverLogger.clone();
        logger.appendTag("remoteAddr", () -> String.format("remote: %s:%d", remoteAddress, remotePort));
    }

    @Override
    public void run() {
        logger.logOnStdout("worker handling socket connection");
        PrintWriter writer;
        BufferedReader reader;
        try {
            writer = new PrintWriter(socket.getOutputStream());
            reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
        } catch (IOException e) {
            logger.logOnStderr("failed to create io from socket");
            throw new RuntimeException(e);
        }

        String msg;
        try {
            while (true) {
                msg = reader.readLine();
                if (msg == null){
                    Thread.sleep(1);
                }else {
                    logger.logOnStdout("msg received: %s", (msg.length() > SHOW_MESSAGE_LENGTH ? msg.substring(0, SHOW_MESSAGE_LENGTH) + " ...(total " + msg.length() + " characters)" : msg));
                    if (msg.equals("close"))
                        break;

                    JSONObject jsonRes = new JSONObject();
                    JSONArray jsonExceptions = new JSONArray();

                    try {
                        JSONObject jsonMsg = JSONObject.parseObject(msg);
                        String dbName = jsonMsg.getString("database");
                        if (dbName == null)
                            dbName = "tsdb";
                        Database database = ArcadeTSDBServer.getOrCreateDatabase(dbName);

                        String action = jsonMsg.getString("action");
                        if (action == null)
                            throw new InvalidMessageException("field 'action' not found");
                        switch (action){
                            case "insert" -> {
                                // get rules
                                JSONObject rules = jsonMsg.getJSONObject("rules");
                                HashMap<String, TSUpdateStrategy> strategyMap= new HashMap<>();
                                if (rules != null){
                                    for (Map.Entry<String, Object> entry : rules.entrySet()){
                                        String metric = entry.getKey();
                                        if (entry.getValue() instanceof JSONObject jsonRule){
                                            String ifExist = jsonRule.getString("ifExist");
                                            if (ifExist == null)
                                                ifExist = "ignore";

                                            String separator = jsonRule.getString("separator");
                                            if (separator == null)
                                                separator = "+";

                                            TSUpdateStrategy strategy = switch (ifExist){
                                                case "error" -> TSUpdateStrategy.ERROR;
                                                case "ignore" -> TSUpdateStrategy.IGNORE;
                                                case "update" -> TSUpdateStrategy.UPDATE;
                                                case "append" -> new TSUpdateStrategy(TSUpdateStrategy.TSBaseUpdateStrategy.APPEND, separator);
                                                default -> throw new InvalidMessageException("cannot resolve rule with ifExist '" + ifExist + "'");
                                            };

                                            strategyMap.put(metric, strategy);
                                        }
                                    }
                                }

                                JSONArray inserts = jsonMsg.getJSONArray("inserts");
                                if (inserts == null)
                                    throw new InvalidMessageException("field 'inserts' not found");
                                handleInsert(database, inserts, strategyMap, jsonRes, jsonExceptions);
                            }
                            case "query" -> {
                                JSONArray queries = jsonMsg.getJSONArray("queries");
                                if (queries == null)
                                    throw new InvalidMessageException("field 'query' not found");
                                handleQuery(database, queries, jsonRes, jsonExceptions);
                            }
                            default -> throw new InvalidMessageException("invalid action '"+action+"'");
                        }
                    }catch (Exception e){
                        logger.logOnStderr(ExceptionSerializer.serializeAll(e));
                        JSONObject jsonException = new JSONObject();
                        jsonException.put("class", e.getClass().getName());
                        jsonException.put("message", e.getMessage());
                        jsonExceptions.add(jsonException);
                    }

                    if (!jsonExceptions.isEmpty())
                        jsonRes.put("exceptions", jsonExceptions);

                    writer.write(jsonRes.toJSONString()+"\n");
                    writer.flush();
                }
            }
        } catch (IOException e) {
            logger.logOnStderr("failed to read msg from client");
        } catch (InterruptedException e) {
            logger.logOnStderr("thread failed when waiting msg");
        }
        writer.write("close\n");
        writer.flush();

        try {
            reader.close();
            writer.close();
            socket.close();
        } catch (IOException ignored) {
        }

        logger.logOnStdout("connection closed");
    }

    private void handleInsert(Database database, JSONArray insertList, Map<String, TSUpdateStrategy> strategyMap, JSONObject jsonResult, JSONArray exceptionList){
        for (int index = 0; index < insertList.size(); index++){
            String retKey = null;
            try {
                JSONObject insert = insertList.getJSONObject(index);
                if (insert == null)
                    throw new InvalidMessageException("cannot resolve insert #"+index);

                String objectId = insert.getString("objectid");
                if (objectId == null)
                    throw new InvalidMessageException("object id not found at insert #"+index);

                retKey = insert.getString("as");
                if (retKey == null){
                    retKey = objectId;
                }
                if (jsonResult.containsKey(retKey)) {
                    String message = "duplicated return key found: " + retKey;
                    retKey = null;
                    throw new InvalidMessageException(message);
                }

                TimeseriesEngine tsEngine = TimeseriesEngine.getInstance(database);

                // start transaction
                synchronized (database) {
                    tsEngine.begin();

                    try {
                        // get object
                        MutableVertex object;
                        try {
                            object = getOrNewVertexById(database, TYPE_OBJECT, PROP_OBJECT_ID, objectId).modify();
                        } catch (CommandSQLParsingException e) {
                            throw new IllegalObjectIdException("illegal objectId " + objectId);
                        }

                        JSONObject props = insert.getJSONObject("props");
                        if (props != null) {
                            for (Map.Entry<String, Object> property : props.entrySet()) {
                                String key = property.getKey();
                                if (key.equals(PROP_OBJECT_ID))
                                    throw new InvalidMessageException("cannot use preserved property " + PROP_OBJECT_ID);
                                object.set(key, property.getValue());
                            }
                            object.save();
                        }

                        JSONArray timeseries = insert.getJSONArray("timeseries");
                        if (timeseries != null) {
                            for (int tsIndex = 0; tsIndex < timeseries.size(); tsIndex++) {
                                JSONObject datapoint = timeseries.getJSONObject(tsIndex);
                                if (datapoint == null)
                                    throw new InvalidMessageException("cannot resolve timeseries #" + tsIndex);

                                Long timestamp = datapoint.getLong("time");
                                if (timestamp == null)
                                    throw new InvalidMessageException("missing timestamp at timeseries #" + tsIndex);

                                Boolean updateIfExist = datapoint.getBoolean("updateIfExist");
                                if (updateIfExist == null)
                                    updateIfExist = false;

                                JSONObject metrics = datapoint.getJSONObject("metrics");
                                if (metrics != null) {
                                    for (Map.Entry<String, Object> metric : metrics.entrySet()) {
                                        String key = metric.getKey();
                                        Object value = metric.getValue();
                                        DataPoint dataPoint;
                                        DataType dataType;
                                        if (value instanceof Number numVal){
                                            if (value instanceof BigDecimal){
                                                dataType = DataType.DOUBLE;
                                                dataPoint = new DoubleDataPoint(timestamp, numVal.doubleValue());
                                            }else{
                                                dataType = DataType.LONG;
                                                dataPoint = new LongDataPoint(timestamp, numVal.longValue());
                                            }
                                        } else if (value instanceof String strVal) {
                                            dataType = DataType.STRING;
                                            dataPoint = new StringDataPoint(timestamp, strVal);
                                        } else {
                                            throw new InvalidMessageException("unsupported data type of " + key + " at timeseries #" + tsIndex);
                                        }

                                        try {
                                            TSUpdateStrategy strategy = strategyMap.get(key);
                                            if (strategy == null)
                                                strategy = TSUpdateStrategy.IGNORE;
                                            tsEngine.insertDataPoint(object, key, dataType, dataPoint, strategy);
                                        } catch (TimeseriesException e){
                                            Constructor<? extends Exception> constructor;
                                            try{
                                                constructor = e.getClass().getConstructor(String.class);
                                            }catch (Exception ignored){
                                                throw e;
                                            }
                                            throw constructor.newInstance(String.format("%s:%s@%d: %s", objectId, key, timestamp, e.getMessage()));
                                        }
                                    }
                                }
                            }
                        }

                        tsEngine.commit();
                    }catch (Exception e){
                        tsEngine.rollback();
                        throw e;
                    }
                }

                JSONObject jsonSubResult = new JSONObject();
                jsonSubResult.put("success", true);
                jsonResult.put(retKey, jsonSubResult);
            }catch (Exception e){
                logger.logOnStderr(ExceptionSerializer.serializeAll(e));
                JSONObject jsonException = new JSONObject();
                jsonException.put("class", e.getClass().getName());
                jsonException.put("message", e.getMessage());

                if (retKey != null){
                    JSONObject jsonSubResult = new JSONObject();
                    jsonSubResult.put("success", false);
                    jsonSubResult.put("exception", jsonException);
                    jsonResult.put(retKey, jsonSubResult);
                }else {
                    exceptionList.add(jsonException);
                }
            }
        }
    }

    public void handleQuery(Database database, JSONArray queryList, JSONObject jsonResult, JSONArray exceptionList){
        for (int index = 0; index < queryList.size(); index++){
            String retKey = null;
            try{
                JSONObject query = queryList.getJSONObject(index);
                if (query == null)
                    throw new InvalidMessageException("cannot resolve query #"+index);

                // object id
                String objectId = query.getString("objectid");
                if (objectId == null)
                    throw new InvalidMessageException("object id not found at query #"+index);

                // query type
                String type = query.getString("type");
                if (type == null)
                    throw new InvalidMessageException("query type not provided at query #"+index);

                // metric
                // if this query requires metric name
                boolean isMetricQuery = METRIC_QUERY_TYPE.contains(type);
                String metric = query.getString("metric");
                if (isMetricQuery && metric == null)
                    throw new InvalidMessageException("metric not found for query type '"+type+"' at query #"+index);

                // return key
                retKey = query.getString("as");
                if (retKey == null){
                    if (isMetricQuery)
                        retKey = objectId+"."+metric+"."+type;
                    else
                        retKey = objectId+"."+type;
                }
                if (jsonResult.containsKey(retKey)) {
                    String message = "duplicated return key found: " + retKey;
                    retKey = null;
                    throw new InvalidMessageException(message);
                }

                TimeseriesEngine tsEngine = TimeseriesEngine.getInstance(database);
                JSONObject jsonSubResult = new JSONObject();
                jsonSubResult.put("success", true);

                // start transaction
                synchronized (database){
                    tsEngine.begin();

                    try{
                        // get object
                        Vertex object;
                        try {
                            object = getVertexById(database, TYPE_OBJECT, PROP_OBJECT_ID, objectId);
                        } catch (CommandSQLParsingException e) {
                            throw new IllegalObjectIdException("illegal objectId " + objectId);
                        }
                        if (object == null) {
                            throw new TargetNotFoundException("object with id '"+objectId+"' not found");
                        }

                        if (type.equals("props")) {
                            JSONObject properties = new JSONObject();
                            for (Map.Entry<String, Object> property : object.toMap(false).entrySet()){
                                String key = property.getKey();
                                if (!key.equals(PROP_OBJECT_ID)){
                                    properties.put(key, property.getValue());
                                }
                            }
                            jsonSubResult.put("result", properties);
                        }else if (type.equals("metrics")){
                            JSONObject metrics = new JSONObject();
                            for (String metricName : tsEngine.getAllMetrics(object)){
                                Statistics stats = tsEngine.aggregativeQuery(object, metricName, 0, Long.MAX_VALUE);
                                if (stats.count == 0) continue;
                                if (stats instanceof FixedStatistics fixedStatistics){
                                    metrics.put(metricName, fixedStatistics.getLastValue());
                                }else{
                                    DataPointSet lastDPSet = tsEngine.periodQuery(object, metricName, stats.lastTime, stats.lastTime);
                                    if (lastDPSet.hasNext()){
                                        metrics.put(metricName, lastDPSet.next().getValue());
                                    }else{
                                        throw new TimeseriesException("last timestamp exist but data point not found at metric '"+metricName+"'");
                                    }
                                }
                            }
                            jsonSubResult.put("result", metrics);
                        }else{
                            // collect timeseries query params
                            // start time
                            Long startTime = query.getLong("start");
                            if (startTime == null)
                                startTime = 0L;
                            if (startTime < 0)
                                throw new InvalidMessageException("start time should be 0 or positive");

                            // end time
                            Long endTime = query.getLong("end");
                            if (endTime == null)
                                endTime = Long.MAX_VALUE;
                            if (endTime < startTime)
                                throw new InvalidMessageException("start time "+startTime+" larger than end time "+endTime);

                            switch (type){
                                case "all" ->{
                                    Long limit = query.getLong("limit");
                                    if (limit == null || limit > 1000)
                                        limit = 1000L;
                                    if (limit <= 0)
                                        throw new InvalidMessageException("limit should be positive");

                                    JSONArray dataPointList = new JSONArray();
                                    DataPointSet dataSet = tsEngine.periodQuery(object, metric, startTime, endTime);
                                    int count = 0;
                                    while (count < limit && dataSet.hasNext()){
                                        DataPoint nextDP = dataSet.next();
                                        JSONObject jsonDataPoint = new JSONObject();
                                        jsonDataPoint.put("time", nextDP.timestamp);
                                        jsonDataPoint.put("value", nextDP.getValue());
                                        dataPointList.add(jsonDataPoint);
                                        count++;
                                    }
                                    jsonSubResult.put("result", dataPointList);
                                }
                                case "count" ->{
                                    Statistics stats = tsEngine.aggregativeQuery(object, metric, startTime, endTime);
                                    jsonSubResult.put("result", stats.count);
                                }
                                case "first" ->{
                                    Statistics stats = tsEngine.aggregativeQuery(object, metric, startTime, endTime);
                                    if (stats.count == 0) {
                                        jsonSubResult.put("result", null);
                                        break;
                                    }

                                    JSONObject jsonDataPoint = new JSONObject();
                                    long firstTime = stats.firstTime;
                                    jsonDataPoint.put("time", firstTime);
                                    if (stats instanceof FixedStatistics fixedStats)
                                        jsonDataPoint.put("value", fixedStats.getFirstValue());
                                    else{
                                        DataPointSet firstDPSet = tsEngine.periodQuery(object, metric, firstTime, firstTime);
                                        if (firstDPSet.hasNext()){
                                            jsonDataPoint.put("value", firstDPSet.next().getValue());
                                        }else{
                                            throw new TimeseriesException("first timestamp exist but data point not found");
                                        }
                                    }
                                    jsonSubResult.put("result", jsonDataPoint);
                                }
                                case "last" ->{
                                    Statistics stats = tsEngine.aggregativeQuery(object, metric, startTime, endTime);
                                    if (stats.count == 0) {
                                        jsonSubResult.put("result", null);
                                        break;
                                    }

                                    JSONObject jsonDataPoint = new JSONObject();
                                    long lastTime = stats.lastTime;
                                    jsonDataPoint.put("time", lastTime);
                                    if (stats instanceof FixedStatistics fixedStats)
                                        jsonDataPoint.put("value", fixedStats.getLastValue());
                                    else{
                                        DataPointSet lastDPSet = tsEngine.periodQuery(object, metric, lastTime, lastTime);
                                        if (lastDPSet.hasNext()){
                                            jsonDataPoint.put("value", lastDPSet.next().getValue());
                                        }else{
                                            throw new TimeseriesException("last timestamp exist but data point not found");
                                        }
                                    }
                                    jsonSubResult.put("result", jsonDataPoint);
                                }
                                case "sum" ->{
                                    Statistics stats = tsEngine.aggregativeQuery(object, metric, startTime, endTime);
                                    if (stats instanceof NumericStatistics numStats){
                                        jsonSubResult.put("result", numStats.getSum());
                                    }else{
                                        throw new DataTypeDismatchException("metric '"+metric+"' is not numeric");
                                    }
                                }
                                case "avg" ->{
                                    Statistics stats = tsEngine.aggregativeQuery(object, metric, startTime, endTime);
                                    if (stats instanceof NumericStatistics numStats){
                                        jsonSubResult.put("result", numStats.getSum().doubleValue()/stats.count);
                                    }else{
                                        throw new DataTypeDismatchException("metric '"+metric+"' is not numeric");
                                    }
                                }
                                case "max" ->{
                                    Statistics stats = tsEngine.aggregativeQuery(object, metric, startTime, endTime);
                                    if (stats instanceof NumericStatistics numStats){
                                        jsonSubResult.put("result", numStats.getMaxValue());
                                    }else{
                                        throw new DataTypeDismatchException("metric '"+metric+"' is not numeric");
                                    }
                                }
                                case "min" ->{
                                    Statistics stats = tsEngine.aggregativeQuery(object, metric, startTime, endTime);
                                    if (stats instanceof NumericStatistics numStats){
                                        jsonSubResult.put("result", numStats.getMinValue());
                                    }else{
                                        throw new DataTypeDismatchException("metric '"+metric+"' is not numeric");
                                    }
                                }
                                default -> throw new InvalidMessageException("unknown query type '"+type+"'");
                            }
                        }

                        tsEngine.commit();
                    }catch (Exception e){
                        tsEngine.rollback();
                        throw e;
                    }
                }

                jsonResult.put(retKey, jsonSubResult);
            }catch (Exception e){
                logger.logOnStderr(ExceptionSerializer.serializeAll(e));
                JSONObject jsonException = new JSONObject();
                jsonException.put("class", e.getClass().getName());
                jsonException.put("message", e.getMessage());

                if (retKey != null){
                    JSONObject jsonSubResult = new JSONObject();
                    jsonSubResult.put("success", false);
                    jsonSubResult.put("exception", jsonException);
                    jsonResult.put(retKey, jsonSubResult);
                }else {
                    exceptionList.add(jsonException);
                }
            }
        }
    }
}
